/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */

package com.kunteng.exchangeboard.mqtt.api;

import com.kunteng.exchangeboard.mqtt.basic.CMD;
import com.kunteng.exchangeboard.mqtt.basic.ICMDCallback;
import net.sf.json.JSONObject;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author Administrator
 */
public class MQTTClient implements MqttCallback {

    static Logger logger = LoggerFactory.getLogger(MQTTClient.class);

    private static MQTTClient instance = null;

    private final Map<CMD, ICMDCallback> callbackMap = new HashMap<CMD, ICMDCallback>();
    private final Map<String, CMD> cmdIDMap = new HashMap<String, CMD>();

    String broker = "tcp://15bi580924.imwork.net:1883";
    String clientId = "YunACClient";

    MemoryPersistence persistence = null;

    MqttClient sampleClient = null;
    ExecutorService threadPool = Executors.newFixedThreadPool(1000);

    int count = 0;

    public static MQTTClient getInstance() {
        if (instance == null) {
            try {
                instance = new MQTTClient();
            } catch (MqttException ex) {
                logger.error("{MQTTServer for MQTTClient DOWN!!!RESTART}");
            }
        }
        return instance;
    }

    private MQTTClient() throws MqttException {
        persistence = new MemoryPersistence();
        clientId = clientId + UUID.randomUUID().toString();
        sampleClient = new MqttClient(broker, clientId, persistence);

        connect();
    }

    private void connect() throws MqttException {
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setCleanSession(true);
        connOpts.setKeepAliveInterval(1000);
        connOpts.setConnectionTimeout(3000);
        sampleClient.setCallback(this);
        sampleClient.connect(connOpts);

        subscribe("/KTPDPOE/#");
    }

    public void send(CMD cmd, ICMDCallback callback) throws MqttException {
//        System.out.println(CMD.toJSON(cmd));
        System.out.println("send CMD:"+cmd.getCmd()+"\t:" + CMD.toPString(cmd));
        MqttMessage message = new MqttMessage(CMD.toPString(cmd).getBytes());
        message.setQos(0);
        callbackMap.put(cmd, callback);
        //此发送消息没有id，使用mac+method区分
        String key = cmd.getDID() + "-" + cmd.getCmd();
        cmdIDMap.put(key, cmd);
        sampleClient.publish(cmd.getTopic(), message);
    }

    public void send(String topoc, String msg) throws MqttException {
        MqttMessage message = new MqttMessage(msg.getBytes());
        sampleClient.publish(topoc, message);
    }

    public Map<CMD, ICMDCallback> getCallbackMap() {
        return callbackMap;
    }

    public Map<String, CMD> getCmdIDMap() {
        return cmdIDMap;
    }

    public void subscribe(String topic) throws MqttException {
        System.out.println("订阅频道：" + topic);
        sampleClient.subscribe(topic, 0);
    }

    public void unsubscribe(String topic) throws MqttException {
        sampleClient.unsubscribe(topic);
    }

    public void connectionLost(Throwable cause) {
        logger.info("{isConnected:" + sampleClient.isConnected() + "}");
        while (!sampleClient.isConnected()) {
            boolean flag = true;
            try {
                System.out.println("正在重新连接...");
                connect();
                flag = false;
            } catch (MqttException e) {
                e.printStackTrace();
            }
            if (flag) {
                try {
                    Thread.sleep(30000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        logger.info("{isConnected:" + sampleClient.isConnected() + "}");
    }

    public void messageArrived(String topic, MqttMessage mm) throws Exception {

        String content = new String(mm.getPayload());
//        System.out.println("MessageArrived:>>>" + topic + "\t" + content);
        if (topic.startsWith("/KTPDPOE")) {
            String mac = topic.split("/")[2];
            String method = topic.split("/")[3];
//            System.out.println("MAC:" + mac + ",method:" + method);
            final JSONObject response = JSONObject.fromObject(content);
            if (method.startsWith("Up_")) {
                method = method.replace("Up_", "Down_");
                final String key = mac + "-" + method;
//                System.out.println("key:"+key);
                final Map<String, CMD> tmpMap = MQTTClient.getInstance().getCmdIDMap();
                if (tmpMap.containsKey(key)) {
                    threadPool.execute(new Runnable() {
                        public void run() {
                            try {
                                MQTTClient.getInstance().getCallbackMap().get(tmpMap.get(key)).process(tmpMap.get(key), response);
                            } catch (Throwable ex) {
                            } finally {
                                MQTTClient.getInstance().getCallbackMap().remove(tmpMap.get(key));
                                tmpMap.remove(key);
                            }
                        }
                    });
                }
            }
        }
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        logger.info("Complate!");
    }

    public static void main(String[] args) {
        MQTTClient.getInstance();
    }

}
